package com.example.demo.controller;

import com.example.demo.componet.ReadCSV;
import com.example.demo.model.BrandJson;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@RestController
@RequestMapping("/analyze")
public class AgeStageController {

    @Resource
    private JavaSparkContext sc;

    //文件地址
    @Autowired
    private String getFilePath;


    private List<ArrayList> getData(){
        ReadCSV readCSV = new ReadCSV();
        List<ArrayList> maps = readCSV.readTwoColumn(getFilePath, 7, 9);    //"C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\test.csv"
        return maps;

    }


//    ReadCSV readCSV = new ReadCSV();
//    List<ArrayList> maps = readCSV.readTwoColumn(getFilePath, 7, 9);    //"C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\test.csv"

    List<Integer> one = new ArrayList<>() ;  //少年 0-18
    List<Integer> two = new ArrayList<>();   // 青年 19-29
    List<Integer> three = new ArrayList<>();  //30-55
    List<Integer> four = new ArrayList<>();   //50 - 70



    @CrossOrigin
    @GetMapping("/chance")
    public List<BrandJson> method1(){
        //建全局的会累加数据

        List<BrandJson> brandJsons =new ArrayList<>();

        for (Object obj:this.getData()
        ) {
            ArrayList a = (ArrayList) obj;
            //年龄
            Integer i = Integer.parseInt((String) a.get(1)) ;
            //消费
            Float f = Float.parseFloat((String) a.get(0)) ;

            if ( 0< i.intValue() && i.intValue()<= 18){
                one.add(f.intValue());
            }
            if (19 <= i.intValue() && i.intValue() <=29){
                two.add(f.intValue());
            }
            if (30<= i.intValue() && i.intValue() <=55){
                three.add(f.intValue());
            }
            if (56<= i.intValue() && i.intValue() <=70){
                four.add(f.intValue());
            }


        }
        //
        one.add(0);one.add(0);two.add(0);two.add(0);three.add(0);three.add(0);four.add(0);four.add(0);

        //RDD

        JavaRDD<Integer> rdd = sc.parallelize(one);
        JavaRDD<Integer> rdd1 = sc.parallelize(two);
        JavaRDD<Integer> rdd2 = sc.parallelize(three);
        JavaRDD<Integer> rdd3 = sc.parallelize(four);

        Integer reduce = rdd.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer reduce1 = rdd1.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer reduce2 = rdd2.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer reduce3 = rdd3.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);

        BrandJson brandJson = new BrandJson();brandJson.setName("少年");brandJson.setValue(reduce);
        BrandJson brandJson1 = new BrandJson();brandJson1.setName("青年");brandJson1.setValue(reduce1);
        BrandJson brandJson2 = new BrandJson();brandJson2.setName("中年");brandJson2.setValue(reduce2);
        BrandJson brandJson3 = new BrandJson();brandJson3.setName("老年");brandJson3.setValue(reduce3);

        brandJsons.add(brandJson);brandJsons.add(brandJson1);brandJsons.add(brandJson2);brandJsons.add(brandJson3);

        return brandJsons;
    }
}
